package com.burges.net.assembly.gelly

import com.burges.net.assembly.gelly.VertexCentricIterations.{getLongLongEdges, getLongLongVertices}
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.graph.{Edge, Vertex}
import org.apache.flink.graph.scala.Graph
import org.apache.flink.graph.spargel.{GatherFunction, MessageIterator, ScatterFunction, ScatterGatherConfiguration}

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/2/26 
  * 描述     Scatter-Gather Iterations实现最短优化路径实例
  */
object ScatterGatherIterations {

	def main(args: Array[String]): Unit = {
		val env = ExecutionEnvironment.getExecutionEnvironment

		// 初始化Graph
		val graph: Graph[Long, Long, Long] = Graph.fromCollection(getLongLongVertices, getLongLongEdges, env)
		val config = new ScatterGatherConfiguration
		//定义最大迭代次数
		val maxIterations = 10
		//执行scatter-gather iteration
		val result = graph.runScatterGatherIteration( new MinDistanceMessager, new VertexDistanceUpdater, maxIterations, config)
		//获取计算结果
		val shortesPaths = result.getVertices

		//定义ScatterFunction
		final class MinDistanceMessager extends ScatterFunction[Long, Long, Long, Long]{
			override def sendMessages(vertex: Vertex[Long, Long]): Unit = {
				for(edge: Edge[Long, Long] <- getEdges){
					sendMessageTo(edge.getTarget, vertex.getValue + edge.getValue)
				}
			}
		}

		//定义GatherFunction更新顶点指标
		final class VertexDistanceUpdater extends GatherFunction[Long, Long, Long]{
			override def updateVertex(vertex: Vertex[Long, Long], messageIterator: MessageIterator[Long]): Unit = {
				var minDistance = Long.MaxValue
				while(messageIterator.hasNext){
					val msg = messageIterator.next()
					if(msg < minDistance){
						minDistance = msg
					}
				}
				//如果顶点的value大于minDistance，更新节点的value
				if(vertex.getValue > minDistance){
					setNewVertexValue(minDistance)
				}
			}
		}
	}

}
